package com.coolingme.producer.asynchronous;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

/**
 * 生产者 发送消息Demo
 *
 * @author wangyue
 * @date 2021/7/9 11:07
 */
@Slf4j
public class ProducerDemo {

    public static void main(String[] args) {
        // 配置文件
        Properties properties = new Properties();
        // 服务地址
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.112.128:9092");
        // 重试次数
        properties.put(ProducerConfig.RETRIES_CONFIG, 1);
        // 1ms
        properties.put(ProducerConfig.LINGER_MS_CONFIG, 1);
        // ACK应答级别
        properties.put(ProducerConfig.ACKS_CONFIG, "all");
        // 批次大小,16KB
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        // 缓冲区大小，32MB
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        // 序列化
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

        // kafka生产者
        KafkaProducer<String, Object> producer = new KafkaProducer<>(properties);
        // Topic名称
        String topicName = "test-topic";

        log.info("============开始==========");

        try {
            for (int i = 0; i < 10; i++) {
                // 发送的数据key
                String key = "record-" + i;
                // 发送的数据value
                String value = "这是一条数据-" + i;
                // 发送的数据对象
                ProducerRecord<String, Object> record = new ProducerRecord<>(topicName, key, value);
                // 发送
                producer.send(record);
                log.info("发送  " + value);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 关闭
            producer.flush();
            producer.close();
        }

    }
}
